package com.hujf.flink.selfFunction;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.IterativeDataSet;
import org.apache.flink.api.java.operators.MapOperator;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.IterativeStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * @Author hujf
 * @Date 2022/5/23 10:28
 * @Description TODO
 */
public class 全量迭代 {
    public static void main(String[] args) throws Exception {
        ExecutionEnvironment environment = ExecutionEnvironment.getExecutionEnvironment();
        IterativeDataSet<Integer> iterate = environment.fromElements(0).iterate(100000000);

        MapOperator<Integer, Integer> map = iterate.map(new MapFunction<Integer, Integer>() {
            @Override
            public Integer map(Integer value) throws Exception {
                double x = Math.random();
                double y = Math.random();

                return value + ((x * x + y * y) < 1 ? 1 : 0);
            }
        });

        DataSet<Integer> integerDataSet = iterate.closeWith(map);

        integerDataSet.map(new MapFunction<Integer, Double>() {
            @Override
            public Double map(Integer value) throws Exception {
                return value/(double)100000000*4;
            }
        }).print();

    }
}
